package com.atguigu.gmallrealtime.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.common.Constant;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author yhm
 * @create 2023-09-22 15:15
 */
public class HBaseUtil {
    public static Connection hBaseConnection;
    public static CompletableFuture<AsyncConnection> asyncConnection;

    static {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");
        try {
            hBaseConnection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
        asyncConnection = ConnectionFactory.createAsyncConnection();
    }

    public static Connection getConnection() throws IOException {
        if (hBaseConnection != null) {
            return hBaseConnection;
        }
        return null;
    }

    public static AsyncConnection getAsyncConnection() throws ExecutionException, InterruptedException {
        return asyncConnection.get();
    }

    public static void closeConnection() throws IOException {
        if (hBaseConnection != null && !hBaseConnection.isClosed()) {
            hBaseConnection.close();
        }
    }
    
    public static void closeAsyncConnection(AsyncConnection asyncConnection) throws IOException {
        if (asyncConnection != null && !asyncConnection.isClosed()) {
            asyncConnection.close();
        }
    }
    

    /**
     * 使用单例hbase连接创建表格
     *
     * @param namespace      命名空间
     * @param tableName      表名
     * @param columnFamilies 列族 可变参数
     * @throws IOException 连接异常
     */
    public static void createTable(String namespace, String tableName, String... columnFamilies) throws IOException {

        // 提前判断列族是否存在
        if (columnFamilies == null || columnFamilies.length == 0) {
            System.out.println("hbase建表至少有一个列族");
            return;
        }

        // 1. 获取admin对象
        Admin admin = hBaseConnection.getAdmin();

        if (admin.tableExists(TableName.valueOf(namespace, tableName))) {
            System.out.println("想要创建的hbase表格已经存在" + namespace + ":" + tableName);
            return;
        }

        // 2. 创建表格描述
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(namespace, tableName));
        for (String columnFamily : columnFamilies) {
            ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
            tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
        }
        // 3. 调用方法创建表格
        try {
            admin.createTable(tableDescriptorBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 4. 关闭连接 释放资源
        admin.close();

    }

    /**
     * 删除表格
     *
     * @param namespace 命名空间
     * @param tableName 表名
     * @throws IOException 连接异常
     */
    public static void dropTable(String namespace, String tableName) throws IOException {
        // 1. 获取admin
        Admin admin = hBaseConnection.getAdmin();
        // 2. 删除表格
        TableName tableName1 = TableName.valueOf(namespace, tableName);
        try {
            admin.disableTable(tableName1);
            admin.deleteTable(tableName1);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 3. 关闭连接 释放资源
        admin.close();
    }

    public static void dropColumn(String namespace, String sinkTable, String rowKey, String sinkFamily, Set<String> columns) throws IOException {
        // 1. 获取table
        Table table = hBaseConnection.getTable(TableName.valueOf(namespace, sinkTable));

        // 2. 创建delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        // 3. 添加列
        for (String column : columns) {
            delete.addColumns(Bytes.toBytes(sinkFamily), Bytes.toBytes(column));
        }

        // 4. 调用table的方式删除数据
        try {
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("删除的数据已经不存在了");
        }

        // 5. 关闭连接 释放资源
        table.close();
    }

    public static void putCell(String namespace, String sinkTable, String rowKey, String sinkFamily, JSONObject jsonObj) throws IOException {
//        if (value == null){
//            value = "";
//        }

        // 1. 获取table对象
        Table table = hBaseConnection.getTable(TableName.valueOf(namespace, sinkTable));

        // 2. 创建put对象
        Put put = new Put(Bytes.toBytes(rowKey));

        // 3. 添加数据
        for (String column : jsonObj.keySet()) {
            String value = jsonObj.getString(column);
            value = value == null ? "" : value;
            put.addColumn(Bytes.toBytes(sinkFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        }

        // 4. 调用table的方法put
        try {
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 5. 关闭连接 释放资源
        table.close();
    }

    public static String getBaseDicDDL() {
        return "CREATE TABLE base_dic (\n" +
                " rowkey STRING,\n" +
                " info ROW<dic_name STRING>,\n" +
                " PRIMARY KEY (rowkey) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'hbase-2.2',\n" +
                " 'table-name' = 'gmall:dim_base_dic',\n" +
                " 'lookup.cache' = 'PARTIAL',\n" +
                " 'lookup.partial-cache.max-rows' = '100',\n" +
                " 'zookeeper.quorum' = '" + Constant.ZOOKEEPER_URL + "'\n" +
                ")";
    }

    /**
     * 直接去hbase中读取维度数据
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public static JSONObject getDim(String tableName, String rowKey) throws IOException {
        Table table = hBaseConnection.getTable(TableName.valueOf("gmall", tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        JSONObject jsonObject = new JSONObject();
        Result result = null;
        try {
            result = table.get(get);
            Cell[] cells = result.rawCells();
            if (cells.length == 0) {
                System.out.println("当前维度表中没有对应的数据" + tableName + "-" + rowKey);
            }
            for (Cell cell : cells) {
                String columnName = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell));
                jsonObject.put(columnName, value);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        table.close();
        return jsonObject;
    }

    /**
     * 添加redis旁路缓存读取数据
     * @param jedis
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public static JSONObject getDimWithRedis(Jedis jedis, String tableName, String rowKey) throws IOException {
        JSONObject jsonObject;
        String key = tableName + ":" + rowKey;
        // 优先读取redis
        String jsonStr = jedis.get(key);

        if (jsonStr == null || jsonStr.length() == 0) {
            // redis里面没有对应的数据
            jsonObject = getDim(tableName, rowKey);
            // 能够读取到数据
            if (jsonObject.keySet().size() != 0) {
                jedis.setex(key, 24 * 60 * 60,jsonObject.toJSONString());
            }
        }else{
            // redis中有数据
            jsonObject = JSON.parseObject(jsonStr);
        }
        return jsonObject;
    }



    public static JSONObject getDimWithRedisAsync(AsyncConnection asyncConnection, RedisAsyncCommands<String, String> async, String tableName, String rowKey) throws IOException, ExecutionException, InterruptedException {
        JSONObject jsonObject;
        String key = tableName + ":" + rowKey;
        // 优先读取redis
        String jsonStr = null;
        try {
            jsonStr = async.get(key).get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        if (jsonStr == null || jsonStr.length() == 0) {
            // redis里面没有对应的数据
            jsonObject = getDimAsync(asyncConnection,tableName, rowKey);
            // 能够读取到数据
            if (jsonObject.keySet().size() != 0) {
                async.setex(key, 24 * 60 * 60,jsonObject.toJSONString());
            }
        }else{
            // redis中有数据
            jsonObject = JSON.parseObject(jsonStr);
        }
        return jsonObject;
    }

    public static JSONObject getDimAsync(AsyncConnection asyncConnection, String tableName, String rowKey) throws ExecutionException, InterruptedException {
        JSONObject jsonObject = new JSONObject();
        AsyncTable<AdvancedScanResultConsumer> table = asyncConnection.getTable(TableName.valueOf("gmall", tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get).get();
        for (Cell cell : result.rawCells()) {
            jsonObject.put(new String(CellUtil.cloneQualifier(cell)),new String(CellUtil.cloneValue(cell)));
        }
        return jsonObject;
    }

}
